package com.garden.rocketmq.learn.samples.config.consumer;

import com.garden.rocketmq.learn.samples.config.condition.NormalConsumerCondition;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * 普通消费者配置类 用于配置普通消费者Bean
 * Created by Garden on 2020-09-17 21:57
 */
@Configuration
public class NormalConsumerConfig {

    public static final Logger LOGGER = LoggerFactory.getLogger(NormalConsumerConfig.class);

    @Value("${rocketmq.consumer.normal.groupName}")
    private String groupName;
    @Value("${rocketmq.consumer.normal.namesrvAdrr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.normal.topics}")
    private String topics;
    @Value("${rocketmq.consumer.normal.consumeThreadMin}")
    private Integer consumeThreadMin;
    @Value("${rocketmq.consumer.normal.consumeThreadMax}")
    private Integer consumeThreadMax;
    @Value("${rocketmq.consumer.normal.consumeMessageBatchMaxSize}")
    private Integer consumeMessageBatchMaxSize;

    @Bean(name = "normalConsumer")
    @Conditional(NormalConsumerCondition.class)
    public DefaultMQPushConsumer normalConsumer() throws RuntimeException{

        //构建普通实时消费者，设置属性
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(this.groupName);
          //设置NameServer的地址，Producer启动后通过NameServer获取目标Topic的路由信息
        defaultMQPushConsumer.setNamesrvAddr(this.namesrvAddr);
          //设置消息实时监听器，用于监听订阅消息，一般用于处理业务逻辑，常独立一个监听类实现
        defaultMQPushConsumer.setMessageListener(new MessageListenerConcurrently(){
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                LOGGER.info("Receive New Messages: thread={},msgs={}", Thread.currentThread().getName(), msgs);
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
          //设置最小线程数以及最大线程数（实时消费者每次消费消息都使用独立的线程）
        defaultMQPushConsumer.setConsumeThreadMin(this.consumeThreadMin);
        defaultMQPushConsumer.setConsumeThreadMax(this.consumeThreadMax);
          //设置一次消费消息的条数，默认为1条
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(this.consumeMessageBatchMaxSize);
          //设置消费方式：
          // 消息模型（广播或集群，默认为集群）；
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
          // 消费起始位置，设置第一次启动时消费消费队列的队首位置（若非第一次启动，将按照上次消费的位置继续消费）
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        try {
              //设置订阅主题以及对应Tags
            defaultMQPushConsumer.subscribe(this.topics,"*");

            //启动实例
            defaultMQPushConsumer.start();
            LOGGER.info("normal consumer start success... groupName={},namesrvAdrr={},topics={}",this.groupName,this.namesrvAddr,this.topics);
        } catch (MQClientException e) {
            LOGGER.error("normal consumer start failed... errorMsg={}",e.getErrorMessage());
            throw new RuntimeException(e);
        }

        //返回Bean
        return defaultMQPushConsumer;
    }


}
